from log import AppLog
from twisted.internet import reactor, protocol, threads, defer
from base_protobuff.base_proto_pb2 import *
from config import *
import subprocess
import json
from twisted.internet.task import LoopingCall
import glob
from datetime import datetime


# twisted框架核心类：协议类
class Protocol(protocol.Protocol):
    def __init__(self):
        self.logger = AppLog.get_logger("Network.Protocol")
        self.buffer = b""  # 处理TCP粘包
        self.name = None
        self.set_id = None

    """---------------协议类 必须实现的函数--------------------"""

    # 新客户端连接时调用
    def connectionMade(self):
        self.logger.info("新的客户端连接: %s", self.transport.getPeer())

    # 收到消息时调用
    def dataReceived(self, data):
        self.buffer += data
        while len(self.buffer) >= HEADER_LEN:
            # 解析协议头 (CMD:4字节 + 长度:4字节)
            cmd = int.from_bytes(self.buffer[:CMD_LEN], byteorder=byteorder_big)
            msg_length = int.from_bytes(
                self.buffer[CMD_LEN:HEADER_LEN],  # 从第4字节开始取4字节
                byteorder=byteorder_big
            )
            # self.logger.debug(f"解析结果 - cmd:{cmd} msg_length:{msg_length}")

            # 检查数据完整性
            total_packet_len = HEADER_LEN + msg_length
            if len(self.buffer) < total_packet_len:
                break

            # 提取协议体
            raw_data = self.buffer[HEADER_LEN:total_packet_len]
            self.buffer = self.buffer[total_packet_len:]

            # 处理协议
            if cmd == CMD_REQ:
                self.handle_protobuf(raw_data)
            else:
                self.logger.warning("未知CMD: %d", cmd)

    # 客户端断开连接时调用
    def connectionLost(self, reason):
        self.logger.debug(f"reason.value: {reason.value}")
        self.logger.info("客户端断开连接: %s", self.transport.getPeer())
        self.factory.clients.remove(self)
        # 从name_map中移除
        self.remove_name_map()
        # 从set_id_map中移除
        self.remove_set_id_map()
        # 打印当前连接数量
        self.logger.info("当前连接数量: %d", len(self.factory.clients))

    """---------------基础收发包函数 -----------------"""

    def send_client(self, cmd, proto_obj):
        body = proto_obj.SerializeToString()
        header = len(body).to_bytes(BODY_LEN, byteorder=byteorder_big)
        cmd_byte = cmd.to_bytes(CMD_LEN, byteorder=byteorder_big)
        self.transport.write(cmd_byte + header + body)
        # self.logger.debug("已发送: %s", proto_obj)

    def send_allclient(self, cmd, proto_obj):
        body = proto_obj.SerializeToString()
        header = len(body).to_bytes(BODY_LEN, byteorder=byteorder_big)
        cmd_byte = cmd.to_bytes(CMD_LEN, byteorder=byteorder_big)
        for client in self.factory.clients:
            client.transport.write(cmd_byte + header + body)
        self.logger.info("已广播消息: %s", proto_obj)

    """向指定name的用户发送消息"""

    def send_to_set_id(self, set_id, cmd, proto_obj):
        body = proto_obj.SerializeToString()
        header = len(body).to_bytes(BODY_LEN, byteorder=byteorder_big)
        cmd_byte = cmd.to_bytes(CMD_LEN, byteorder=byteorder_big)
        protocols = self.factory.get_connections(set_id)
        for protocol in protocols:
            try:
                # 增强类型和状态检查
                if not isinstance(protocol, Protocol):
                    self.logger.warning("非法连接对象类型: %s (应为Protocol实例)", type(protocol))
                    self.factory.set_id_map[set_id]["connections"] = []
                    self.factory.set_id_map[set_id]["svr_role_num"] = 0
                    return

                if not hasattr(protocol, 'transport'):
                    self.logger.warning("协议对象缺少transport属性: %s", protocol)
                    self.factory.set_id_map[set_id]["connections"] = []
                    self.factory.set_id_map[set_id]["svr_role_num"] = 0
                    return

                transport = protocol.transport
                if transport is None:
                    self.logger.warning("连接[%s] transport为空", protocol.name)
                    self.factory.set_id_map[set_id]["connections"] = []
                    self.factory.set_id_map[set_id]["svr_role_num"] = 0
                    return

                if not transport.connected:
                    self.logger.warning("连接[%s] 已断开", protocol.name)
                    self.factory.set_id_map[set_id]["connections"] = []
                    self.factory.set_id_map[set_id]["svr_role_num"] = 0
                    return

                # 发送消息
                transport.write(cmd_byte + header + body)
                # self.logger.debug("成功发送至[%s]", protocol.name)

            except Exception as e:
                self.logger.error("发送给[%s]时发生异常: %s",
                                  getattr(protocol, 'name', '未知连接'),
                                  str(e),
                                  exc_info=True)

    """---------------分发处理函数 --------------------"""

    def handle_protobuf(self, data):
        # self.logger.debug("收到原始数据: %s", data)
        try:
            req = REQ()
            req.ParseFromString(data)
            # self.logger.debug("解析后的REQ: %s", req)

            if req.type == TRANS_TYPE_LOGIN:
                self.login(req)
            elif req.type == TRANS_TYPE_CHECK_SVR:
                self.check_svr(req)
            elif req.type == TRANS_TYPE_GET_SVR_DATA:
                self.get_svr_data(req)
            elif req.type == TRANS_TYPE_LOCK_SVR:
                self.lock_svr(req)
            elif req.type == TRANS_TYPE_AUTOMAKE:
                self.auto_make(req)
            elif req.type == TRANS_TYPE_KILL_CFG:
                self.kill_cfg(req)
            elif req.type == TRANS_TYPE_RELOAD:
                self.reload(req)
            elif req.type == TRANS_TYPE_STOP:
                self.stop(req)
            elif req.type == TRANS_TYPE_DEL_SVR:
                self.del_svr(req)
            elif req.type == TRANS_TYPE_SYN:
                self.syn(req)
            elif req.type == TRANS_TYPE_DUMP:
                self.dump(req)
            elif req.type == TRANS_TYPE_SET_SVR_TIME:
                self.set_svr_time(req)
            elif req.type == TRANS_TYPE_ASK:
                self.ask(req)
            elif req.type == TRANS_TYPE_SVN_INFO:
                self.get_svn_info(req)
            elif req.type == TRANS_TYPE_PUBLIC:
                self.auto_make_public(req)
            elif req.type == TRANS_TYPE_KILL_GAME:
                self.kill_game(req)
            elif req.type == TRANS_TYPE_ENTER:
                self.enter(req)
            elif req.type == TRANS_TYPE_CHANGE_SET_ID:
                self.change_set_id(req)
            elif req.type == TRANS_TYPE_CHANGE_GM_STATE:
                self.change_gm_state(req)
            else:
                self.logger.warning("未知请求类型: %d", req.type)

        except Exception as e:
            self.logger.error("协议解析失败: %s", e, exc_info=True)

    """---------------工具函数 --------------------"""

    # 获取区服运行状态
    def get_set_run_state(self, set_id):
        set_data = self.factory.set_id_map.get(set_id)
        return set_data["run_state"] if set_data else 0

    # 获取当前操作者
    def get_cur_name(self, set_id):
        set_data = self.factory.set_id_map.get(set_id)
        return set_data["cur_name"] if set_data else ""

    def get_task_name(self, set_id):
        set_data = self.factory.set_id_map.get(set_id)
        return set_data["task_name"] if set_data else ""

    def get_lock_name(self, set_id):
        set_data = self.factory.set_id_map.get(set_id)
        return set_data["lock_name"] if set_data else ""

    def get_svr_role_num(self, set_id):
        set_data = self.factory.set_id_map.get(set_id)
        return set_data["svr_role_num"] if set_data else 0

    def get_update_time(self, set_id):
        self.create_main_server_data(set_id)
        set_data = self.factory.set_id_map.get(set_id)
        return set_data["update_time"] if set_data else 0

    # 增加name_map
    def add_name_map(self):
        self.factory.name_map[self.name] = self

    # 删除name_map
    def remove_name_map(self):
        if self.name in self.factory.name_map:
            del self.factory.name_map[self.name]

    # 增加set_id_map
    def add_set_id_map(self):
        # 初始化区服状态字典（如果不存在）
        if self.set_id not in self.factory.set_id_map:
            self.logger.info("找不到区服状态，初始化一个新的")
            self.factory.set_id_map[self.set_id] = {
                "run_state": 0,  # 0:空闲，1:运行中
                "cur_name": "",  # 当前操作者
                "task_name": "",  # 任务名称
                "lock_name": "",  # 锁定者
                "connections": [],  # 当前区服的客户端连接
                "svr_role_num": 0,
                "update_time": 0
            }

        if isinstance(self, Protocol):
            self.factory.set_id_map[self.set_id]["connections"].append(self)
            self.factory.set_id_map[self.set_id]["svr_role_num"] += 1
        else:
            self.logger.error("尝试添加非Protocol对象到connections")

    # 删除set_id_map
    def remove_set_id_map(self):
        if hasattr(self, 'set_id') and self.set_id in self.factory.set_id_map:
            set_data = self.factory.set_id_map[self.set_id]
            if "connections" in set_data and self in set_data["connections"]:
                set_data["connections"].remove(self)
                set_data["svr_role_num"] -= 1
                if set_data["svr_role_num"] > 0:
                    self.send_set_data_ntf(self.set_id)

    def send_set_data_ntf(self, tmp_set_id):
        run_state = self.get_set_run_state(tmp_set_id)
        cur_name = self.get_cur_name(tmp_set_id)
        task_name = self.get_task_name(tmp_set_id)
        lock_name = self.get_lock_name(tmp_set_id)
        svr_role_num = self.get_svr_role_num(tmp_set_id)
        update_time = self.get_update_time(tmp_set_id)
        main_update_time = self.get_update_time(MAIN_SERVER_ID)
        self.send_to_set_id(tmp_set_id, CMD_NTF, NTF(set_id=tmp_set_id,
                                                     run_state=run_state,
                                                     cur_name=cur_name,
                                                     task_name=task_name,
                                                     lock_name=lock_name,
                                                     svr_role_num=svr_role_num,
                                                     update_time=update_time,
                                                     main_update_time=main_update_time
                                                    ))

    def set_set_data(self, tmp_set_id, tmp_run_state=None, tmp_cur_name=None, tmp_task_name=None, tmp_lock_name=None,
                     tmp_update_time=None):
        set_data = self.factory.set_id_map.get(tmp_set_id)
        if set_data is not None:
            self.logger.info(f"set_set_data 1: {set_data}")

            if tmp_run_state is not None:
                set_data["run_state"] = tmp_run_state
            if tmp_cur_name is not None:
                set_data["cur_name"] = tmp_cur_name
            if tmp_task_name is not None:
                set_data["task_name"] = tmp_task_name
            if tmp_lock_name is not None:
                set_data["lock_name"] = tmp_lock_name
            if tmp_update_time is not None:
                set_data["update_time"] = tmp_update_time

            self.logger.info(f"set_set_data 2: {set_data}")

    def set_and_ntf_set_data(self, tmp_set_id, tmp_run_state=None, tmp_cur_name=None, tmp_task_name=None,
                             tmp_lock_name=None, tmp_update_time=None):
        set_data = self.factory.set_id_map.get(tmp_set_id)
        if set_data is not None:
            self.set_set_data(tmp_set_id, tmp_run_state, tmp_cur_name, tmp_task_name, tmp_lock_name, tmp_update_time)
            self.send_set_data_ntf(tmp_set_id)

    """ -----------------------新增辅助方法---------------------------"""
    """异步执行make debug"""

    def _async_make_debug(self, path):
        d = threads.deferToThread(subprocess.run,
                                  ['make', 'debug=1'],
                                  cwd=path,
                                  stdout=subprocess.PIPE,
                                  stderr=subprocess.PIPE,
                                  check=True)
        d.addCallback(lambda _: self.logger.info(f"make debug成功: {path}"))
        return d

    """异步执行make clean"""

    def _async_make_clean(self, path):
        d = threads.deferToThread(subprocess.run,
                                  ['make', 'clean'],
                                  cwd=path,
                                  stdout=subprocess.PIPE,
                                  stderr=subprocess.PIPE,
                                  check=True)
        d.addCallback(lambda _: self.logger.info(f"make clean成功: {path}"))
        return d

    """异步执行SVN更新"""

    def _async_svn_update(self, path):
        d = threads.deferToThread(subprocess.run,
                                  ['svn', 'update'],
                                  cwd=path,
                                  stdout=subprocess.PIPE,
                                  stderr=subprocess.PIPE,
                                  check=True)
        d.addCallback(lambda _: self.logger.info(f"SVN更新成功: {path}"))
        return d

    """异步等待并检查崩溃文件"""

    def _async_wait_and_check(self, svr_id, name, wait_time):
        def _check_core_files():
            RUN_PATH = f'/home/qie{svr_id}/code/run'
            core_files = glob.glob(os.path.join(RUN_PATH, '**', 'core*'), recursive=True)
            if core_files:
                self.logger.info(f"检测到{svr_id}服务器崩溃文件")
                self._show_status(svr_id, name, '检测到崩溃文件')
                self.send_client(CMD_RSP, RSP(result=RET_FAIL_10006, name=self.name, set_id=req.set_id, type=req.type))
                self.set_and_ntf_set_data(svr_id, STATE_STOP, NONE_NAME, NONE_NAME)
                return True

        d = defer.Deferred()
        reactor.callLater(wait_time, lambda:
        threads.deferToThread(_check_core_files).chainDeferred(d))
        return False

    def _async_check_public(self, svr_id, name, wait_time):
        def _check_core_files():
            run_path = f'/home/qie_cwar_1/code/run'
            core_files = glob.glob(os.path.join(run_path, '**', 'core*'), recursive=True)
            if core_files:
                self.logger.info(f"检测到{run_path}服务器崩溃文件")
                self._show_status(svr_id, name, '检测到崩溃文件')
                self.send_client(CMD_RSP, RSP(result=RET_FAIL_10006, name=self.name, set_id=req.set_id, type=req.type))
                self.set_and_ntf_set_data(svr_id, STATE_STOP, NONE_NAME, NONE_NAME)
                return True
            run_path = f'/home/qie_cross/code/run'
            core_files = glob.glob(os.path.join(run_path, '**', 'core*'), recursive=True)
            if core_files:
                self.logger.info(f"检测到{run_path}服务器崩溃文件")
                self._show_status(svr_id, name, '检测到崩溃文件')
                self.send_client(CMD_RSP, RSP(result=RET_FAIL_10006, name=self.name, set_id=req.set_id, type=req.type))
                self.set_and_ntf_set_data(svr_id, STATE_STOP, NONE_NAME, NONE_NAME)
                return True
            run_path = f'/home/qie_global/code/run'
            core_files = glob.glob(os.path.join(run_path, '**', 'core*'), recursive=True)
            if core_files:
                self.logger.info(f"检测到{run_path}服务器崩溃文件")
                self._show_status(svr_id, name, '检测到崩溃文件')
                self.send_client(CMD_RSP, RSP(result=RET_FAIL_10006, name=self.name, set_id=req.set_id, type=req.type))
                self.set_and_ntf_set_data(svr_id, STATE_STOP, NONE_NAME, NONE_NAME)
                return True
            run_path = f'/home/qie_proxy/code/run'
            core_files = glob.glob(os.path.join(run_path, '**', 'core*'), recursive=True)
            if core_files:
                self.logger.info(f"检测到{run_path}服务器崩溃文件")
                self._show_status(svr_id, name, '检测到崩溃文件')
                self.send_client(CMD_RSP, RSP(result=RET_FAIL_10006, name=self.name, set_id=req.set_id, type=req.type))
                self.set_and_ntf_set_data(svr_id, STATE_STOP, NONE_NAME, NONE_NAME)
                return True

        d = defer.Deferred()
        reactor.callLater(wait_time, lambda:
        threads.deferToThread(_check_core_files).chainDeferred(d))
        return False

    """显示状态信息"""

    # 企业微信机器人发送消息
    def _show_status(self, svr_id, name, message):
        threads.deferToThread(subprocess.run, ['bash', 'show.sh', str(svr_id), name, message], cwd=SHOW_SCRIPT_DIR)

    def _handle_process_error(self, error, svr_id, name):
        """处理进程错误"""
        err_msg = f"命令执行失败: {error.cmd}\n错误: {error.stderr.decode().strip()}"
        self.logger.error(err_msg)
        self._show_status(svr_id, name, '执行错误！')
        raise  # 传递给上层异常处理

    def _handle_generic_error(self, error, svr_id, name):
        """处理通用错误"""
        err_msg = f"处理过程中发生意外错误: {str(error)}"
        self.logger.error(err_msg, exc_info=True)
        self._show_status(svr_id, name, '服务器在处理过程中发生意外错误！')
        raise

    """---------------功能处理 --------------------"""

    # 创建主干服的数据
    def create_main_server_data(self, set_id):
        set_data = self.factory.set_id_map.get(set_id)
        if set_data is None:
            self.factory.set_id_map[set_id] = {
                "run_state": 0,
                "cur_name": "",
                "task_name": "",
                "lock_name": "",
                "connections": [],
                "svr_role_num": 0,
                "update_time": 0
            }
            self.logger.info(f"创建区服数据: {set_id}")

    def check_common(self, set_id):
        set_data = self.factory.set_id_map.get(set_id)
        if set_data is None:
            return RET_FAIL_10001
        if set_data["run_state"] == 1:
            return RET_FAIL_10002
        return RET_SUCCESS

    def login(self, req):
        self.logger.info("----------------登录请求 - 参数: %s", req)

        if req.param_1 != VERSION:
            self.send_client(CMD_RSP, RSP(result=RET_FAIL_10004,
                                          name=req.name,
                                          type=req.type))
            return
        else:
            # 保存用户信息到当前实例
            self.set_id = req.set_id
            self.add_set_id_map()
            self.send_client(CMD_RSP, RSP(result=RET_SUCCESS,
                                          name=req.name,
                                          set_id=req.set_id,
                                          type=req.type))

    """检查区服进程状态"""

    def check_svr(self, req):
        # self.logger.info("----------------区服进程状态请求 - 参数: %s", req)
        try:
            # 同步执行命令
            command = f"pgrep ^{req.set_id}gamesvrd$"
            # self.logger.debug(f"执行命令: {command}")
            ret = subprocess.run(
                command,
                shell=True,
                stdout=subprocess.PIPE,
                stderr=subprocess.PIPE
            )

            # 构造响应
            # self.logger.debug(f"执行结果: {ret.returncode}")
            tmp_ret = RET_FAIL
            if ret.returncode == 0:
                tmp_ret = RET_SUCCESS
            self.send_client(CMD_RSP, RSP(result=tmp_ret,
                                          name=req.name,
                                          type=req.type))
        except Exception as e:
            self.logger.debug(f"检查失败: {str(e)}")
            self.send_client(CMD_RSP, RSP(result=RET_FAIL,
                                          name=req.name,
                                          type=req.type))

    def get_svr_data(self, req):
        # self.logger.info("----------------获取区服数据请求 - 参数: %s", req)
        self.send_client(CMD_RSP, RSP(result=RET_SUCCESS, name=req.name, set_id=req.set_id, type=req.type))
        self.send_set_data_ntf(req.set_id)

    def lock_svr(self, req):
        self.logger.info("----------------锁定请求 - 参数: %s", req)
        set_data = self.factory.set_id_map.get(req.set_id)
        print(f"set_data: {set_data}")
        ret = RET_SUCCESS
        if set_data is None:
            ret = RET_FAIL
        else:
            ret = RET_SUCCESS
            if req.param_1 == 0:
                set_data["lock_name"] = ""
                self.logger.info("区服已解锁: %s", req.set_id)
            else:
                set_data["lock_name"] = req.name
                self.logger.info("区服已锁定: %s", req.set_id)
            self.send_set_data_ntf(req.set_id)
        self.send_client(CMD_RSP, RSP(result=ret,
                                      name=req.name,
                                      set_id=req.set_id,
                                      type=req.type,
                                      param_1=req.param_1))

    def auto_make(self, req):
        self.logger.info("----------------自动编译请求 - 参数: %s", req)

        @defer.inlineCallbacks
        def async_auto_make():
            try:
                svr_id = req.set_id
                name = req.name

                # 添加初始状态日志
                self.logger.info(f"[{name}] 开始处理服务器 {svr_id} 的自动编译请求 1")

                # 检查区服状态
                ret = self.check_common(svr_id)
                if ret != RET_SUCCESS:
                    self.send_client(CMD_RSP, RSP(result=ret, name=name, set_id=svr_id, type=req.type))
                    return

                if svr_id in server_data['主干服']:
                    ret = self.check_common(MAIN_SERVER_ID)
                    if ret != RET_SUCCESS:
                        self.send_client(CMD_RSP, RSP(result=ret, name=name, set_id=svr_id, type=req.type))
                        return

                # 设置运行状态
                cur_time = int(datetime.now().timestamp())
                self.logger.info(f"[{name}] 设置服务器 {svr_id} 为运行状态 2")
                self.set_and_ntf_set_data(svr_id, STATE_RUN, name, "自动编译", None, cur_time)

                self.logger.info(f"[{name}] 编译设置服务器[主干服]为运行状态 2.1")
                self.set_set_data(MAIN_SERVER_ID, STATE_RUN, name, "自动编译", None, cur_time)

                # 异步执行编译任务
                try:
                    self.logger.info(f"[{name}] 开始执行自动编译任务 2.0")
                    self._show_status(svr_id, name, '编译开始')

                    # 普通服务器处理
                    yield self._async_svn_update(cfg_path)
                    yield self._async_svn_update(src_path)
                    yield self._async_make_clean(src_path)
                    yield self._async_make_debug(src_path)

                    self.logger.info(f"[{name}] 编译取消服务器[主干服]为运行状态 2.5")
                    self.set_set_data(MAIN_SERVER_ID, STATE_STOP, NONE_NAME, NONE_NAME)

                    # 检查崩溃日志文件
                    command = f"cat {src_path}/build_log.txt | grep error"
                    # self.logger.info(f"[{name}] 执行命令: {command}")
                    ret = subprocess.run(
                        command,
                        shell=True,
                        stdout=subprocess.PIPE,
                        stderr=subprocess.PIPE
                    )

                    # 如果找到匹配的内容，退出状态码为 0
                    if ret.returncode == 0:
                        self._show_status(svr_id, name, '编译失败')
                        self.send_client(CMD_RSP, RSP(result=RET_FAIL_10005, name=self.name, set_id=req.set_id, type=req.type))
                        self.set_and_ntf_set_data(svr_id, STATE_STOP, NONE_NAME, NONE_NAME)
                        self.logger.info(f"[{name}] 编译失败")
                        return

                    yield threads.deferToThread(subprocess.run, ['bash', 'syn_vension.sh', str(svr_id)],
                                                cwd=BASH_SCRIPT_DIR)

                    yield threads.deferToThread(subprocess.run, ['bash', 'onekillserver.sh', str(svr_id)],
                                                cwd=BASH_SCRIPT_DIR)

                    run_path = f'/home/qie{svr_id}/code/run/'
                    yield threads.deferToThread(subprocess.run, ['bash', 'start_qie_game.sh'], cwd=run_path)

                    self.logger.info(f"[{name}] 等待并检查崩溃文件 5")
                    core_ret = yield self._async_wait_and_check(svr_id, name, 15)
                    if core_ret:
                        return

                    # 成功处理
                    self._show_status(svr_id, name, '编译结束')
                    self.send_client(CMD_RSP, RSP(result=RET_SUCCESS, name=self.name, set_id=req.set_id, type=req.type))
                    self.set_and_ntf_set_data(svr_id, STATE_STOP, NONE_NAME, NONE_NAME)

                except subprocess.CalledProcessError as e:
                    self._handle_process_error(e, svr_id, name)
                except Exception as e:
                    self._handle_generic_error(e, svr_id, name)

            finally:
                # 最终状态清理
                self.logger.info(f"[{req.name}] 自动编译请求 处理完成 7")

        # 立即启动异步任务
        reactor.callWhenRunning(async_auto_make)

    def kill_cfg(self, req):
        self.logger.info("----------------热更配置请求 - 参数: %s", req)

        @defer.inlineCallbacks
        def async_kill_cfg():
            try:
                # 参数校验和状态检查
                svr_id = req.set_id
                name = req.name

                # 检查区服状态
                self.logger.info(f"[{name}] 开始处理服务器 {svr_id} 的热更配置请求 1")
                ret = self.check_common(svr_id)
                if ret != RET_SUCCESS:
                    self.send_client(CMD_RSP, RSP(result=ret, name=name, set_id=svr_id, type=req.type))
                    return

                self.logger.info(f"[{name}] 设置服务器 {svr_id} 为运行状态 2")
                self.set_and_ntf_set_data(svr_id, STATE_RUN, name, "热更配置")
                try:
                    self._show_status(svr_id, name, '热更开始')
                    # 异步执行SVN更新
                    self.logger.info(f"[{name}] 开始执行SVN更新 3")
                    yield self._async_svn_update(cfg_path)
                    # 异步执行kill命令
                    self.logger.info(f"[{name}] 开始执行kill命令 5")
                    yield threads.deferToThread(subprocess.run, ['bash', 'sys_kill_cfg.sh', str(svr_id)],
                                                cwd=BASH_SCRIPT_DIR, check=True)

                    # 异步等待并检查崩溃文件
                    self.logger.info(f"[{name}] 开始等待并检查崩溃文件 6")
                    core_ret = yield self._async_wait_and_check(svr_id, name, 15)
                    if core_ret:
                        return

                    # 成功处理
                    self.logger.info(f"[{name}] 处理完成 7")
                    self._show_status(svr_id, name, '热更结束')
                    self.send_client(CMD_RSP, RSP(result=RET_SUCCESS, name=self.name, set_id=req.set_id, type=req.type))
                    self.set_and_ntf_set_data(svr_id, STATE_STOP, NONE_NAME, NONE_NAME)

                except subprocess.CalledProcessError as e:
                    self._handle_process_error(e, svr_id, name)
                except Exception as e:
                    self._handle_generic_error(e, svr_id, name)

            finally:
                # 最终状态清理
                self.logger.info(f"[{req.name}] 热更配置 处理完成 8")

        # 立即启动异步任务
        reactor.callWhenRunning(async_kill_cfg)

    def reload(self, req):
        self.logger.info("----------------停服重启 - 参数: %s", req)

        @defer.inlineCallbacks
        def async_reload():
            try:
                # 参数校验和状态检查
                svr_id = req.set_id
                name = req.name

                # 检查区服状态
                self.logger.info(f"[{name}] 停服重启")
                ret = self.check_common(svr_id)
                if ret != RET_SUCCESS:
                    self.send_client(CMD_RSP, RSP(result=ret, name=name, set_id=svr_id, type=req.type))
                    return

                self.set_and_ntf_set_data(svr_id, STATE_RUN, name, "停服重启")

                try:
                    self._show_status(svr_id, name, '开始停服重启')
                    yield threads.deferToThread(subprocess.run, ['bash', 'onekillserver.sh', str(svr_id)],
                                                cwd=BASH_SCRIPT_DIR)
                    run_path = f'/home/qie{svr_id}/code/run/'
                    yield threads.deferToThread(subprocess.run, ['bash', 'start_qie_game.sh'], cwd=run_path)

                    self.logger.info(f"[{name}] 等待并检查崩溃文件")
                    core_ret = yield self._async_wait_and_check(svr_id, name, 15)
                    if core_ret:
                        return

                    self._show_status(svr_id, name, '停服重启完成')
                    self.send_client(CMD_RSP, RSP(result=RET_SUCCESS, name=self.name, set_id=req.set_id, type=req.type))
                    self.set_and_ntf_set_data(svr_id, STATE_STOP, NONE_NAME, NONE_NAME)

                except subprocess.CalledProcessError as e:
                    self._handle_process_error(e, svr_id, name)
                except Exception as e:
                    self._handle_generic_error(e, svr_id, name)

            finally:
                # 最终状态清理
                self.logger.info(f"[{req.name}] 停服重启 处理完成")

        # 立即启动异步任务
        reactor.callWhenRunning(async_reload)

    def stop(self, req):
        self.logger.info("----------------停服请求 - 参数: %s", req)

        @defer.inlineCallbacks
        def async_stop():
            try:
                # 参数校验和状态检查
                svr_id = req.set_id
                name = req.name

                # 检查区服状态
                self.logger.info(f"[{name}] 停服开始")
                ret = self.check_common(svr_id)
                if ret != RET_SUCCESS:
                    self.send_client(CMD_RSP, RSP(result=ret, name=name, set_id=svr_id, type=req.type))
                    return

                self.set_and_ntf_set_data(svr_id, STATE_RUN, name, "停服")

                try:
                    yield threads.deferToThread(subprocess.run, ['bash', 'onekillserver.sh', str(svr_id)],
                                                cwd=BASH_SCRIPT_DIR)

                    self._show_status(svr_id, name, '停服完成')

                    self.logger.info(f"[{name}] 停服完成")
                    self.send_client(CMD_RSP, RSP(result=RET_SUCCESS, name=self.name, set_id=req.set_id, type=req.type))
                    self.set_and_ntf_set_data(svr_id, STATE_STOP, NONE_NAME, NONE_NAME)

                except subprocess.CalledProcessError as e:
                    self._handle_process_error(e, svr_id, name)
                except Exception as e:
                    self._handle_generic_error(e, svr_id, name)

            finally:
                # 最终状态清理
                self.logger.info(f"[{req.name}] 停服完成 处理完成")

        # 立即启动异步任务
        reactor.callWhenRunning(async_stop)

    def del_svr(self, req):
        self.logger.info("----------------删档请求 - 参数: %s", req)

        @defer.inlineCallbacks
        def async_del_svr():
            try:
                # 参数校验和状态检查
                svr_id = req.set_id
                name = req.name

                # 检查区服状态
                ret = self.check_common(svr_id)
                if ret != RET_SUCCESS:
                    self.send_client(CMD_RSP, RSP(result=ret, name=name, set_id=svr_id, type=req.type))
                    return

                self.set_and_ntf_set_data(svr_id, STATE_RUN, name, "删档")

                try:
                    self.logger.info(f"[{name}] 删档开始")
                    self._show_status(svr_id, name, '删档开始')

                    run_path = f'/home/qie{svr_id}/code/run/'
                    yield threads.deferToThread(subprocess.run, ['bash', 'onekillserver.sh', str(svr_id)],
                                                cwd=BASH_SCRIPT_DIR)

                    yield threads.deferToThread(subprocess.run, ['bash', 'reset_pg_db_26.sh', str(svr_id)],
                                                cwd=BASH_SCRIPT_DIR)

                    yield threads.deferToThread(subprocess.run, ['bash', 'start_qie_game.sh'],
                                                cwd=run_path)

                    self.logger.info(f"[{name}] 等待并检查崩溃文件")
                    core_ret = yield self._async_wait_and_check(svr_id, name, 15)
                    if core_ret:
                        return

                    self._show_status(svr_id, name, '删档完成')
                    self.logger.info(f"[{name}] 删档完成")
                    self.send_client(CMD_RSP, RSP(result=RET_SUCCESS, name=self.name, set_id=req.set_id, type=req.type))
                    self.set_and_ntf_set_data(svr_id, STATE_STOP, NONE_NAME, NONE_NAME)

                except subprocess.CalledProcessError as e:
                    self._handle_process_error(e, svr_id, name)
                except Exception as e:
                    self._handle_generic_error(e, svr_id, name)

            finally:
                # 最终状态清理
                self.logger.info(f"[{req.name}] 删档完成 处理完成")

        # 立即启动异步任务
        reactor.callWhenRunning(async_del_svr)

    def syn(self, req):
        self.logger.info("----------------同步请求 - 参数: %s", req)

        @defer.inlineCallbacks
        def async_syn():
            try:
                # 参数校验和状态检查
                svr_id = req.set_id
                name = req.name
                param_1 = int(req.param_1)
                data = 'data'
                if param_1 == 1:
                    data = 'version'

                # 检查区服状态
                ret = self.check_common(svr_id)
                if ret != RET_SUCCESS:
                    self.send_client(CMD_RSP, RSP(result=ret, name=name, set_id=svr_id, type=req.type))
                    return

                cur_time = self.get_update_time(MAIN_SERVER_ID)
                self.set_and_ntf_set_data(svr_id, STATE_RUN, name, "同步", None, cur_time)

                try:
                    self.logger.info(f"[{name}] 开始同步")
                    self._show_status(svr_id, name, '开始同步')

                    yield threads.deferToThread(subprocess.run, ['bash', 'syn_vension.sh', str(svr_id)],
                                                cwd=BASH_SCRIPT_DIR)

                    yield threads.deferToThread(subprocess.run, ['bash', 'onekillserver.sh', str(svr_id)],
                                                cwd=BASH_SCRIPT_DIR)

                    run_path = f'/home/qie{svr_id}/code/run/'
                    yield threads.deferToThread(subprocess.run, ['bash', 'start_qie_game.sh'], cwd=run_path)

                    self.logger.info(f"[{name}] 等待并检查崩溃文件")
                    core_ret = yield self._async_wait_and_check(svr_id, name, 15)
                    if core_ret:
                        return

                    self.logger.info(f"[{name}] 同步完成")
                    self._show_status(svr_id, name, '同步完成')
                    self.send_client(CMD_RSP, RSP(result=RET_SUCCESS, name=self.name, set_id=req.set_id, type=req.type))
                    self.set_and_ntf_set_data(svr_id, STATE_STOP, NONE_NAME, NONE_NAME)

                except subprocess.CalledProcessError as e:
                    self._handle_process_error(e, svr_id, name)
                except Exception as e:
                    self._handle_generic_error(e, svr_id, name)

            finally:
                # 最终状态清理
                self.logger.info(f"[{req.name}] 同步完成 处理完成")

        # 立即启动异步任务
        reactor.callWhenRunning(async_syn)

    def dump(self, req):
        self.logger.info("----------------导号请求 - 参数: %s", req)

    def set_svr_time(self, req):
        self.logger.info("----------------设置服务器时间请求 - 参数: %s", req)

    def ask(self, req):
        self.logger.info("----------------询问请求 - 参数: %s", req)
        self.send_to_set_id(req.set_id, CMD_RSP,
                            RSP(result=RET_SUCCESS, name=self.name, set_id=req.set_id, type=req.type))

    def get_svn_info(self, req):
        self.logger.info("----------------SVN信息请求 - 参数: %s", req)

        @defer.inlineCallbacks
        def async_get_svn_info():
            try:
                svr_id = req.set_id
                name = req.name

                self.logger.debug(f"[{name}] 获取config_SVN信息，路径: {cfg_path}")
                self.logger.debug(f"[{name}] 获取src_SVN信息，路径: {src_path}")

                def get_svn_data(path_type, path):
                    try:
                        result = subprocess.run(
                            ['svn', 'info', path],
                            stdout=subprocess.PIPE,
                            stderr=subprocess.PIPE,
                            universal_newlines=True,
                            check=True
                        )
                        info = {}
                        for line in result.stdout.split('\n'):
                            if ':' in line:
                                key, value = line.split(':', 1)
                                info[key.strip()] = value.strip()
                        return {path_type: info}
                    except Exception as e:
                        self.logger.error(f"{path_type}路径SVN信息获取失败: {str(e)}")
                        return {path_type: {"error": str(e)}}

                # 并行获取配置和代码的SVN信息
                config_defer = threads.deferToThread(get_svn_data, "config", cfg_path)
                src_defer = threads.deferToThread(get_svn_data, "src", src_path)

                results = yield defer.DeferredList([config_defer, src_defer], consumeErrors=True)

                # 解析结果
                svn_data = {}
                for success, result in results:
                    if success:
                        svn_data.update(result)

                """
                config_revision=svn_data.get("config", {}).get("Revision", "0"),
                config_url=svn_data.get("config", {}).get("URL", ""),
                config_last_author=svn_data.get("config", {}).get("Last Changed Author", ""),

                src_revision=svn_data.get("src", {}).get("Revision", "0"),
                src_url=svn_data.get("src", {}).get("URL", ""),
                src_last_author=svn_data.get("src", {}).get("Last Changed Author", "")
                """

                config_url = svn_data.get("config", {}).get("URL", "") or ""
                src_url = svn_data.get("src", {}).get("URL", "") or ""

                self.logger.info("准备发送SVN信息: config=%s | src=%s", config_url, src_url)

                self.send_client(CMD_SVN_INFO_NTF, SVN_INFO_NTF(config_svn_url=config_url, src_svn_url=src_url))
                self.logger.info(f"[{name}] SVN信息获取成功!")
                self.send_client(CMD_RSP, RSP(result=RET_SUCCESS, name=name, set_id=svr_id, type=req.type))

            except Exception as e:
                self.logger.error("处理SVN信息异常: %s", str(e), exc_info=True)

        # 启动异步任务
        reactor.callWhenRunning(async_get_svn_info)

    def auto_make_public(self, req):
        self.logger.info("----------------公共进程更新请求 - 参数: %s", req)
        self.send_to_set_id(req.set_id, CMD_RSP,
                            RSP(result=RET_SUCCESS, name=self.name, set_id=req.set_id, type=req.type))

        @defer.inlineCallbacks
        def async_auto_make_public():
            try:
                svr_id = req.set_id
                name = req.name

                # 添加初始状态日志
                self.logger.info(f"[{name}] 开始处理服务器的公共进程更新请求 1")

                # 检查区服状态
                ret = self.check_common(svr_id)
                if ret != RET_SUCCESS:
                    self.send_client(CMD_RSP, RSP(result=ret, name=name, set_id=svr_id, type=req.type))
                    return

                if svr_id in server_data['主干服']:
                    ret = self.check_common(MAIN_SERVER_ID)
                    if ret != RET_SUCCESS:
                        self.send_client(CMD_RSP, RSP(result=ret, name=name, set_id=svr_id, type=req.type))
                        return

                # 设置运行状态
                cur_time = int(datetime.now().timestamp())
                self.logger.info(f"[{name}] 设置服务器 {svr_id} 为运行状态 2")
                self.set_and_ntf_set_data(svr_id, STATE_RUN, name, "公共进程更新", None, cur_time)

                self.logger.info(f"[{name}] 编译设置服务器[主干服]为运行状态 2.1")
                self.set_set_data(MAIN_SERVER_ID, STATE_RUN, name, "公共进程更新", None, cur_time)

                # 异步执行编译任务
                try:
                    self.logger.info(f"[{name}] 开始执行公共进程更新任务 2.0")
                    self._show_status(svr_id, name, '编译开始')

                    # 普通服务器处理
                    yield self._async_svn_update(cfg_path)
                    yield self._async_svn_update(src_path)
                    yield self._async_make_clean(src_path)
                    yield self._async_make_debug(src_path)

                    self.logger.info(f"[{name}] 编译取消服务器[主干服]为运行状态 2.5")
                    self.set_set_data(MAIN_SERVER_ID, STATE_STOP, NONE_NAME, NONE_NAME)

                    # 检查崩溃日志文件
                    command = f"cat {src_path}/build_log.txt | grep error"
                    # self.logger.info(f"[{name}] 执行命令: {command}")
                    ret = subprocess.run(
                        command,
                        shell=True,
                        stdout=subprocess.PIPE,
                        stderr=subprocess.PIPE
                    )

                    # 如果找到匹配的内容，退出状态码为 0
                    if ret.returncode == 0:
                        self._show_status(svr_id, name, '编译失败')
                        self.send_client(CMD_RSP, RSP(result=RET_FAIL_10005, name=self.name, set_id=req.set_id, type=req.type))
                        self.set_and_ntf_set_data(svr_id, STATE_STOP, NONE_NAME, NONE_NAME)
                        self.logger.info(f"[{name}] 编译失败")
                        return

                    # 同步
                    yield threads.deferToThread(subprocess.run, ['bash', 'syn_vension_p.sh'],
                                                cwd=BASH_SCRIPT_DIR)

                    cwar_path = f'/home/qie_cwar_1/code/run'
                    cross_path = f'/home/qie_cross/code/run'
                    global_path = f'/home/qie_global/code/run'
                    proxy_path = f'/home/qie_proxy/code/run'

                    # 停服
                    yield threads.deferToThread(subprocess.run, ['bash', 'stop_cwar_qie.sh'],
                                                cwd=cwar_path)
                    yield threads.deferToThread(subprocess.run, ['bash', 'stop_cross_qie.sh'],
                                                cwd=cross_path)
                    yield threads.deferToThread(subprocess.run, ['bash', 'stop_global_qie.sh'],
                                                cwd=global_path)
                    yield threads.deferToThread(subprocess.run, ['bash', 'stop_proxy_qie.sh'],
                                                cwd=proxy_path)
                    # 起服
                    yield threads.deferToThread(subprocess.run, ['bash', 'start_cwar_qie.sh'],
                                                cwd=cwar_path)
                    yield threads.deferToThread(subprocess.run, ['bash', 'start_cross_qie.sh'],
                                                cwd=cross_path)
                    yield threads.deferToThread(subprocess.run, ['bash', 'start_global_qie.sh'],
                                                cwd=global_path)
                    yield threads.deferToThread(subprocess.run, ['bash', 'start_proxy_qie.sh'],
                                                cwd=proxy_path)

                    self.logger.info(f"[{name}] 等待并检查崩溃文件 5")
                    core_ret = yield self._async_check_public(svr_id, name, 15)
                    if core_ret:
                        return

                    # 成功处理
                    self._show_status(svr_id, name, '公共进程更新结束')
                    self.send_client(CMD_RSP, RSP(result=RET_SUCCESS, name=self.name, set_id=req.set_id, type=req.type))
                    self.set_and_ntf_set_data(svr_id, STATE_STOP, NONE_NAME, NONE_NAME)

                except subprocess.CalledProcessError as e:
                    self._handle_process_error(e, svr_id, name)
                except Exception as e:
                    self._handle_generic_error(e, svr_id, name)

            finally:
                # 最终状态清理
                self.logger.info(f"[{req.name}] 自动编译请求 处理完成 7")

        # 立即启动异步任务
        reactor.callWhenRunning(async_auto_make_public)

    def kill_game(self , req):
        self.logger.info("----------------kill_game - 参数: %s", req)

        @defer.inlineCallbacks
        def async_kill_game():
            try:
                # 参数校验和状态检查
                svr_id = req.set_id
                name = req.name

                # 检查区服状态
                self.logger.info(f"[{name}] 开始处理服务器 {svr_id} 的kill_game 请求 1")
                ret = self.check_common(svr_id)
                if ret != RET_SUCCESS:
                    self.send_client(CMD_RSP, RSP(result=ret, name=name, set_id=svr_id, type=req.type))
                    return

                self.logger.info(f"[{name}] 设置服务器 {svr_id} 为运行状态 2")
                self.set_and_ntf_set_data(svr_id, STATE_RUN, name, "kill_game")
                try:
                    # self._show_status(svr_id, name, 'kill_game开始')

                    # 异步执行kill命令
                    self.logger.info(f"[{name}] 开始执行kill命令 5")
                    yield threads.deferToThread(subprocess.run, ['bash', 'sys_kill_game.sh', str(svr_id)],
                                                cwd=BASH_SCRIPT_DIR, check=True)

                    # 异步等待并检查崩溃文件
                    self.logger.info(f"[{name}] kill_game 开始等待并检查崩溃文件 6")
                    core_ret = yield self._async_wait_and_check(svr_id, name, 15)
                    if core_ret:
                        return

                    # 成功处理
                    self.logger.info(f"[{name}] 处理完成 7")
                    self._show_status(svr_id, name, 'kill_game完成')
                    self.send_client(CMD_RSP, RSP(result=RET_SUCCESS, name=self.name, set_id=req.set_id, type=req.type))
                    self.set_and_ntf_set_data(svr_id, STATE_STOP, NONE_NAME, NONE_NAME)

                except subprocess.CalledProcessError as e:
                    self._handle_process_error(e, svr_id, name)
                except Exception as e:
                    self._handle_generic_error(e, svr_id, name)

            finally:
                # 最终状态清理
                self.logger.info(f"[{req.name}] 热更配置 处理完成 8")

        # 立即启动异步任务
        reactor.callWhenRunning(async_kill_game)


    def enter(self, req):
        self.logger.info("----------------进入请求，发送服务器列表 - 参数: %s", req)
        self.send_to_set_id(req.set_id, CMD_RSP,
                            RSP(result=RET_SUCCESS, name=self.name, set_id=req.set_id, type=req.type))
        self.name = req.name
        self.add_name_map()
        self.send_server_data()

    def send_server_data(self):
        """发送动态服务器组数据"""
        ntf = SERVER_LIST_NTF()
        # 动态构建服务器组
        for group_name, server_ids in server_data.items():
            server_group = ntf.server_groups.add()
            server_group.group_name = group_name
            server_group.server_ids.extend(server_ids)
        self.send_client(CMD_SERVER_LIST_NTF, ntf)

    def change_set_id(self, req):
        self.logger.info("----------------退出服务器请求 - 参数: %s", req)
        self.remove_set_id_map()
        self.send_to_set_id(req.set_id, CMD_RSP,
                            RSP(result=RET_SUCCESS, name=self.name, set_id=req.set_id, type=req.type))

    def change_gm_state(self, req):
        self.logger.info("----------------切换GM状态请求 - 参数: %s", req)


# twisted框架核心类：工厂类
class Factory(protocol.Factory):
    def __init__(self):
        self.logger = AppLog.get_logger("System.Factory")
        self.clients = set()
        self.name_map = {}  # 新增：name到Protocol实例的映射
        self.set_id_map = {}  # 新增：set_id到Protocol实例列表的映射
        self.logger.info("服务器初始化完成")
        self.mmap_file = "set_data.mmap"
        self.save_time = 600  # 5分钟
        # 启动定时任务
        self.save_task = LoopingCall(self.save_to_mmap)
        self.save_task.start(self.save_time, now=False)

    def get_connections(self, set_id):
        set_data = self.set_id_map.get(set_id)
        return set_data.get("connections", []) if set_data else []

    def buildProtocol(self, addr):
        self.logger.info("新的客户端连接: %s", addr)
        p = Protocol()
        p.factory = self
        self.clients.add(p)
        return p

    def stopFactory(self):
        self.logger.info("服务器关闭中...")
        for client in self.clients:
            client.transport.loseConnection()
        self.logger.info("所有客户端连接已断开")

    def get_set_data(self, set_id):
        """获取区服数据对象（线程安全）"""
        # 通过主线程操作确保线程安全
        return reactor.callFromThread(lambda: self.set_id_map.get(set_id, None))

    """将set_id_map序列化到mmap文件"""

    def save_to_mmap(self):
        # self.logger.info(f"保存mmap数据：{self.set_id_map}")
        data = json.dumps(self.set_id_map, default=self._serialize).encode('utf-8')
        with open(self.mmap_file, "wb") as f:
            f.write(len(data).to_bytes(4, 'big'))  # 写入数据长度
            f.write(data)
        self.logger.info("set_id_map已保存到mmap文件")

    """处理无法JSON序列化的对象（如Protocol实例）"""

    def _serialize(self, obj):

        if isinstance(obj, Protocol):
            return None  # 不保存连接对象
        return obj.__dict__  # 转换其他对象为字典

    """从mmap文件加载数据"""

    def load_from_mmap(self):
        try:
            with open(self.mmap_file, "rb") as f:
                length_bytes = f.read(4)
                if not length_bytes:
                    return
                length = int.from_bytes(length_bytes, 'big')
                data = f.read(length).decode('utf-8')

                # 加载原始数据（此时键为字符串）
                loaded_data = json.loads(data, object_hook=lambda d: self._deserialize(d))

                # 转换键为整数类型
                converted_data = {}
                for key_str in loaded_data:
                    try:
                        set_id = int(key_str)
                        converted_data[set_id] = loaded_data[key_str]
                    except ValueError:
                        self.logger.error(f"无效的区服ID格式: {key_str}，跳过该条目")
                        continue

                # 保留现有连接信息
                for set_id in converted_data:
                    converted_data[set_id]["run_state"] = 0
                    converted_data[set_id]["cur_name"] = ""
                    converted_data[set_id]["task_name"] = ""
                    converted_data[set_id]["connections"] = []
                    converted_data[set_id]["svr_role_num"] = 0

                self.set_id_map = converted_data
                self.logger.info("从mmap文件恢复set_id_map成功")
                self.logger.debug("加载后的完整数据: %s", self.set_id_map)
        except FileNotFoundError:
            self.logger.warning("mmap文件不存在，跳过加载")
        except json.JSONDecodeError as e:
            self.logger.error("JSON解析失败: %s", str(e))
        except Exception as e:
            self.logger.error("加载mmap文件失败: %s", str(e), exc_info=True)

    """反序列化时重建数据结构"""

    def _deserialize(self, dct):
        return dct


if __name__ == "__main__":
    AppLog.get_logger("System.Factory").info("服务器启动")
    AppLog()
    factory = Factory()
    factory.load_from_mmap()
    reactor.listenTCP(PORT, factory)
    reactor.run()
